package com.sunday.common.mq.rocket.brave.study.base.e9_Trace;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.junit.jupiter.api.Test;

import java.util.concurrent.TimeUnit;

/**
 * 消息轨迹
 * 1. 消息轨迹数据关键属性
 * Producer端	Consumer端	Broker端
 * 生产实例信息	消费实例信息 	消息的Topic
 * 发送消息时间	投递时间，投递轮次 	消息存储位置
 * 消息是否发送成功	消息是否消费成功	消息的Key值
 * 发送耗时	消费耗时 	消息的Tag值
 * 2. 支持消息轨迹集群部署
 * 2.1 Broker端配置文件
 * 这里贴出Broker端开启消息轨迹特性的properties配置文件内容：
 * <p>
 * brokerClusterName=DefaultCluster
 * brokerName=broker-a
 * brokerId=0
 * deleteWhen=04
 * fileReservedTime=48
 * brokerRole=ASYNC_MASTER
 * flushDiskType=ASYNC_FLUSH
 * storePathRootDir=/data/rocketmq/rootdir-a-m
 * storePathCommitLog=/data/rocketmq/commitlog-a-m
 * autoCreateSubscriptionGroup=true
 * ## if msg tracing is open,the flag will be true
 * traceTopicEnable=true
 * listenPort=10911
 * brokerIP1=XX.XX.XX.XX1
 * namesrvAddr=XX.XX.XX.XX:9876
 * 2.2 普通模式
 * RocketMQ集群中每一个Broker节点均用于存储Client端收集并发送过来的消息轨迹数据。因此，对于RocketMQ集群中的Broker节点数量并无要求和限制。
 * <p>
 * 2.3 物理IO隔离模式
 * 对于消息轨迹数据量较大的场景，可以在RocketMQ集群中选择其中一个Broker节点专用于存储消息轨迹，使得用户普通的消息数据与消息轨迹数据的物理IO完全隔离，互不影响。在该模式下，RockeMQ集群中至少有两个Broker节点，其中一个Broker节点定义为存储消息轨迹数据的服务端。
 * <p>
 * 2.4 启动开启消息轨迹的Broker
 * nohup sh mqbroker -c ../conf/2m-noslave/broker-a.properties &
 *
 * 3. 保存消息轨迹的Topic定义
 * RocketMQ的消息轨迹特性支持两种存储轨迹数据的方式：
 *
 * 3.1 系统级的TraceTopic
 * 在默认情况下，消息轨迹数据是存储于系统级的TraceTopic中(其名称为：RMQ_SYS_TRACE_TOPIC)。该Topic在Broker节点启动时，会自动创建出来（如上所叙，需要在Broker端的配置文件中将traceTopicEnable的开关变量设置为true）。
 *
 * 3.2 用户自定义的TraceTopic
 * 如果用户不准备将消息轨迹的数据存储于系统级的默认TraceTopic，也可以自己定义并创建用户级的Topic来保存轨迹（即为创建普通的Topic用于保存消息轨迹数据）。下面一节会介绍Client客户端的接口如何支持用户自定义的TraceTopic。
 *
 * 4. 支持消息轨迹的Client客户端实践
 * 为了尽可能地减少用户业务系统使用RocketMQ消息轨迹特性的改造工作量，作者在设计时候采用对原来接口增加一个开关参数(enableMsgTrace)来实现消息轨迹是否开启；并新增一个自定义参数(customizedTraceTopic)来实现用户存储消息轨迹数据至自己创建的用户级Topic。
 *
 */
@Slf4j
public class TraceProducer {

    @Test
    public void producer() throws Exception {

        // 实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer("trace_producer_group_name");
        // 设置NameServer的地址
        producer.setNamesrvAddr("127.0.0.1:9876");
        // 启动Producer实例
        producer.start();

        for (int i = 0; i < 1; i++) {
            // 创建消息，并指定Topic，Tag和消息体
            Message msg = new Message(
                    "TopicTest" /* Topic */,
                    "TagA" /* Tag */,
                    ("Hello RocketMQ ").getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
            );
//            msg.putUserProperty(MessageConst.);

            // 消息在1S内没有发送成功，就会重试
            SendResult sendResult = producer.send(msg, 1000);
            // 通过sendResult返回消息是否成功送达
            log.info("{}", sendResult);
        }

        // 如果不再发送消息，关闭Producer实例。
        producer.shutdown();
    }


}
